package com.huang.canal.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.huang.canal.entity.ESbean;
import com.huang.canal.service.EsServiceimpl;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
public class CanalListener  {


    @Autowired
    private EsServiceimpl serviceimpl;
    @Autowired
    private RestHighLevelClient client;

    private static CanalListener util=new CanalListener();

    @PostConstruct
    public void inte(){
        util.serviceimpl=serviceimpl;
        util.client=client;
    }

    //启动项目执行
    public void run() {
        System.out.println("canal开始监控mysql");
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",
                11111), "example", "canal", "canal");
        connector.connect();
//            connector.subscribe(".*\\..*");
        connector.subscribe("xm.edu_course,xm.edu_publish,xm.edu_teacher");
//        connector.subscribe();
        connector.rollback();
        try {
            while (true) {
                Message message = connector.getWithoutAck(1000);//每次接受一千条数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    System.out.println("数据库发送变化");
                    printEntry(message.getEntries());
                }
                connector.ack(batchId); // 提交确认
            }
        }catch (Exception e) {
            e.printStackTrace();
        } finally {
            connector.disconnect();
        }
    }
    //开始对变化数据进行操作
    private  void printEntry(List<CanalEntry.Entry> entrys) throws IOException {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            CanalEntry.RowChange rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            CanalEntry.EventType eventType = rowChage.getEventType();

            System.out.println(String.format("================ binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));

            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.INSERT) {
                    // 如果是课程表修改了   通过课程id查  更新ES
                    esInsert(rowData.getAfterColumnsList());

                } else if (eventType == CanalEntry.EventType.DELETE) {
                    esDelete(rowData.getBeforeColumnsList());
                } else {//EventType.UPDATE

                    if (entry.getHeader().getTableName().endsWith("course")) {
                        //课程表修改了
                        esUpdate(rowData.getAfterColumnsList(), "course");
                    } else if (entry.getHeader().getTableName().endsWith("teacher")) {
                        esUpdate(rowData.getAfterColumnsList(), "teacher");
                    }

                }
            }
        }
    }

    private  void esUpdate(List<CanalEntry.Column> afterColumnsList, String table) throws IOException {
        boolean isSuccess=false;
        UpdateRequest request=new UpdateRequest();
        request.index("course_db");
        if(table.equals("course")){
            for (CanalEntry.Column column : afterColumnsList) {
                if("id".equals(column.getName())){
                    ESbean one= util.serviceimpl.findOne(column.getValue());
                    request.id(one.getId());
                    //将我们的数据放入请求Json中
                    request.doc(JSON.toJSONString(one), XContentType.JSON);
                    //客户端发送请求,获取相应的结果
                    UpdateResponse response = util.client.update(request,RequestOptions.DEFAULT);
                    isSuccess = response.getShardInfo().getSuccessful() > 0 ? true : false;
                }
                if (isSuccess) break;
            }
        }else{
            for (CanalEntry.Column column : afterColumnsList) {
                UpdateResponse response =null;
                if("id".equals(column.getName())){
                    List<ESbean> list= util.serviceimpl.findOne2(column.getValue());
                    for (ESbean eSbean : list) {
                        request.id(eSbean.getId());
                        //将我们的数据放入请求Json中
                        request.doc(JSON.toJSONString(eSbean), XContentType.JSON);
                        //客户端发送请求,获取相应的结果
                         response = util.client.update(request,RequestOptions.DEFAULT);
                    }

                    isSuccess = response.getShardInfo().getSuccessful() > 0 ? true : false;
                }
                if (isSuccess) break;
            }
        }


    }

    private  void esDelete(List<CanalEntry.Column> afterColumnsList) throws IOException {
        for (CanalEntry.Column column : afterColumnsList) {
            String id = column.getName();
            if ("id".equals(id)) {
                DeleteRequest request = new DeleteRequest("course_db");
                request.id(column.getValue());
                request.timeout("1s");
                DeleteResponse deleteResponse = util.client.delete(request, RequestOptions.DEFAULT);
                System.out.println("删除了");
            }
        }

    }


    private  void esInsert(List<CanalEntry.Column> columns) throws IOException {
        IndexRequest request = new IndexRequest("course_db");
        boolean isSuccess = false;
        //课程表修改了
        for (CanalEntry.Column column : columns) {
            if ("id".equals(column.getName())) {
                ESbean one = util.serviceimpl.findOne(column.getValue());
                request.id(one.getId());
                //将我们的数据放入请求Json中
                request.source(JSON.toJSONString(one), XContentType.JSON);
                //客户端发送请求,获取相应的结果
                IndexResponse response = util.client.index(request, RequestOptions.DEFAULT);
                isSuccess = response.getShardInfo().getSuccessful() > 0 ? true : false;
            }
            if (isSuccess) break;
        }

    }





}